package cn.zwx.learn.basic.mongo.config;

import cn.zwx.learn.basic.mongo.listener.DocumentationMessageListener;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.model.changestream.FullDocument;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import static org.springframework.data.mongodb.core.aggregation.Aggregation.match;
import static org.springframework.data.mongodb.core.aggregation.Aggregation.newAggregation;
import static org.springframework.data.mongodb.core.query.Criteria.where;

/**
 * @description: 配置 mongo监听器的容器MessageListenerContainer，
 * spring启动时会自动启动监听的任务用于接收changestream
 * @projectName:java-basic
 * @see:cn.zwx.learn.basic.mongo.config
 * @author:zhangwenxue
 * @createTime:2021/5/4 22:20
 * @version:1.0
 */
@Configuration
@Component
public class MongoConfig {


    @Value("${spring.data.mongodb-target.uri}")
    private String uri;

    @Value("${spring.data.mongodb-target.database}")
    private String database;

    @Bean("readClient")
    public MongoClient newClient() {
        MongoClient client = MongoClients.create(uri);
        return client;
    }

    @Bean("mongoReadTemplate")
    public MongoTemplate newTemplate(@Autowired@Qualifier("readClient") MongoClient mongoClient) {
        MongoTemplate template = new MongoTemplate(mongoClient,database);
        return template;

    }


    /**
     * @description 注册监听规则
     * @return org.springframework.data.mongodb.core.messaging.MessageListenerContainer
     * @author zhangwenxue
     * @createTime 2021/5/4 22:30
     **/
    @Bean
    public MessageListenerContainer messageListenerContainer(@Autowired@Qualifier("mongoReadTemplate") MongoTemplate template, @Autowired DocumentationMessageListener documentationMessageListener) {
        Executor executor = Executors.newSingleThreadExecutor();
        MessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(template, executor) {
            @Override
            public boolean isAutoStartup() {
                return true;
            }
        };

        ChangeStreamRequest<Document> request = ChangeStreamRequest.builder(documentationMessageListener)
                //需要监听的数据库名称，不指定默认监听所有的数据库
                .database("yun_shop_dev")
                // 监听数据库下面的表
                .collection("t_goods_info")
                //过滤需要监听的操作类型，可以根据需求指定过滤条件
                .filter(newAggregation(match(where("operationType").in("insert", "update", "replace","remove","save","delete"))))
                .fullDocumentLookup(FullDocument.UPDATE_LOOKUP) //不设置时，文档更新时，只会发送变更字段的信息，设置UPDATE_LOOKUP会返回文档的全部信息
                .build();
        messageListenerContainer.register(request, Document.class);

        return messageListenerContainer;
    }

}
